package queue

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/worklz/yunj-blog-go-micro-common/log"
	"gitee.com/worklz/yunj-blog-go-micro-common/redis"
	"gitee.com/worklz/yunj-blog-go-micro-common/util"
	"runtime/debug"
	"sync"
	"time"

	goRedis "github.com/redis/go-redis/v9"
)

type QueueHandleError struct {
	Time    time.Time
	Latency uint64 // 耗时（单位：ms）
	Err     error
}

type QueueLastHandleResult struct {
	Time    time.Time
	Status  int
	Latency uint64 // 耗时（单位：ms）
	Err     error
}

// 队列接口
type QueueInterface interface {
	Name() string                                // 队列名称
	Desc() string                                // 队列描述
	LogPrefix() string                           // 日志前缀
	CoroutineNum() int                           // 开启的协程数量
	MessageNum() int                             // 每次获取消费的消息数量
	Tries() int                                  // 消息处理失败重试次数
	Timeout() time.Duration                      // 消息消费处理超时时间
	DefaultStartStatus() uint8                   // 默认启动状态
	GetStatus() uint8                            // 获取状态
	SetStatus(status uint8) error                // 设置状态
	SetRunAt(at *time.Time)                      // 设置运行时间
	GetRunAt() *time.Time                        // 获取运行时间
	IsRunning() bool                             // 是否处理运行中
	SetPauseAt(at *time.Time)                    // 设置暂停时间
	GetPauseAt() *time.Time                      // 获取暂停时间
	GetHeartbeatTime() *time.Time                // 获取心跳时间
	GetLastHandleResult() *QueueLastHandleResult // 获取最近一次处理结果
	GetHandleErrors() []*QueueHandleError        // 获取近期处理错误（最近10次）
	Start() (err error)                          // 启动
	Stop()                                       // 停止
	Handler([]string)                            // 消息消费处理函数
	Push(message any) error                      // 添加消息
}

// 队列
type Queue struct {
	runWg              sync.WaitGroup
	runChan            chan struct{}
	runMu              sync.Mutex             // 用于保护 runChan 的启动和停止操作
	status             uint8                  // 处理状态。对应Status
	runAt              *time.Time             // 处理运行时间
	pauseAt            *time.Time             // 处理暂停时间
	heartbeatTime      *time.Time             // 处理心跳时间
	heartbeatTimeMu    sync.Mutex             // 用于保护处理心跳时间的设置
	lastHandleResult   *QueueLastHandleResult // 最近一次处理结果
	lastHandleResultMu sync.Mutex             // 用于保护最近一次处理结果的设置
	handleErrors       []*QueueHandleError    // 近期处理错误（最近10次）
	logPrefix          string                 // 日志前缀
	queueInstance      QueueInterface         // 队列实例
}

// 编译时判断是否实现接口
var _ QueueInterface = (*Queue)(nil)

// InitQueueInstance 初始化队列实例
func (q *Queue) InitQueueInstance(queueInstance QueueInterface) {
	q.queueInstance = queueInstance
}

func (q *Queue) Name() string {
	//TODO implement me
	panic("implement me")
}

func (q *Queue) Desc() string {
	//TODO implement me
	panic("implement me")
}

func (q *Queue) Timeout() time.Duration {
	//TODO implement me
	panic("implement me")
}

func (q *Queue) Handler(strings []string) {
	//TODO implement me
	panic("implement me")
}

// 获取状态
func (q *Queue) GetStatus() uint8 {
	return q.status
}

// 设置状态
func (q *Queue) SetStatus(status uint8) error {
	if _, ok := StatusEnum.IsValue(status); !ok {
		return fmt.Errorf("%s status %d is not valid", q.LogPrefix(), status)
	}
	q.status = status
	return nil
}

// 定义默认的启动状态（没有配置数据库状态时应用）
func (q *Queue) DefaultStartStatus() uint8 {
	return STATUS_PAUSED
}

// 获取运行时间
func (q *Queue) GetRunAt() *time.Time {
	return q.runAt
}

// 设置运行时间
func (q *Queue) SetRunAt(at *time.Time) {
	q.runAt = at
	q.pauseAt = nil
}

// 是否处理运行中
func (q *Queue) IsRunning() bool {
	return q.runAt != nil
}

// 获取处理的暂停时间
func (q *Queue) GetPauseAt() *time.Time {
	return q.pauseAt
}

// 设置处理的暂停时间
func (q *Queue) SetPauseAt(at *time.Time) {
	q.pauseAt = at
	q.runAt = nil
}

// 设置心跳时间
func (q *Queue) SetHeartbeatTime(heartbeatTime *time.Time) {
	q.heartbeatTimeMu.Lock()
	defer q.heartbeatTimeMu.Unlock()
	q.heartbeatTime = heartbeatTime
}

// 获取心跳时间
func (q *Queue) GetHeartbeatTime() *time.Time {
	return q.heartbeatTime
}

// 设置最近一次处理结果
func (q *Queue) SetLastHandleResult(res *QueueLastHandleResult) {
	q.lastHandleResultMu.Lock()
	defer q.lastHandleResultMu.Unlock()
	q.lastHandleResult = res
}

// 获取最近一次处理结果
func (q *Queue) GetLastHandleResult() *QueueLastHandleResult {
	return q.lastHandleResult
}

// 获取近期处理错误
func (q *Queue) GetHandleErrors() []*QueueHandleError {
	return q.handleErrors
}

// 开启的协程数量
func (q *Queue) CoroutineNum() int {
	return 1
}

// 每次获取消费的消息数量
func (q *Queue) MessageNum() int {
	return 1
}

// 尝试次数
// 指定消息咋处理异常时的最大尝试次数（默认3次）
func (q *Queue) Tries() int {
	return 3
}

// 日志前缀
func (q *Queue) LogPrefix() string {
	if q.logPrefix != "" {
		return q.logPrefix
	}
	name := q.queueInstance.Name()
	desc := q.queueInstance.Desc()
	if desc != "" || name != "" {
		q.logPrefix = fmt.Sprintf("队列[%s][%s]", name, desc)
	} else {
		q.logPrefix = fmt.Sprintf("队列[%s]", util.PackageFullName(q.queueInstance))
	}
	return q.logPrefix
}

// 队列消息推送
// 示例：queue.New(&queue.Tests{}).Push()
func (q *Queue) Push(message any) error {
	queueName := q.queueInstance.Name()
	if queueName == "" {
		return fmt.Errorf("%s未指定队列（未定义队列名称）", q.LogPrefix())
	}
	if _, isStr := message.(string); !isStr {
		msg, err := util.ToJson(message)
		if err != nil {
			return err
		}
		message = msg
	}
	err := redis.LPush(context.Background(), queueName, message)
	if err != nil {
		return fmt.Errorf("%s消息推送异常！%v", q.LogPrefix(), err)
	}
	return nil
}

// 启动
func (q *Queue) Start() (err error) {
	q.runMu.Lock()
	defer q.runMu.Unlock()
	if q.queueInstance.GetStatus() == STATUS_RUNNING && q.IsRunning() {
		return fmt.Errorf("%s已启动！", q.LogPrefix())
	}
	// 初始化创建处理通道
	q.runChan = make(chan struct{})
	// 开启配置数量协程
	coroutineNum := q.queueInstance.CoroutineNum()
	q.runWg.Add(coroutineNum)
	for i := 1; i <= coroutineNum; i++ {
		go q.runJob(i)
	}
	currTime := time.Now().In(time.Local)
	err = q.SetStatus(STATUS_RUNNING)
	if err != nil {
		return
	}
	q.SetRunAt(&currTime)

	return
}

// 运行队列任务
func (q *Queue) runJob(i int) {
	var messages []string
	defer func() {
		if err := recover(); err != nil {
			// 记录异常日志
			q.LogError(fmt.Sprintf("协程[%d]panic:", i), err)
			// 判断是否已经获取到消息了，重新入队
			q.rePushMessages(messages, false)
			// 停止调度
			return
		}
		q.runWg.Done()
	}()
	q.LogInfo(fmt.Sprintf("协程[%d]开始调度！", i))
	for {
		currTime := time.Now().In(time.Local)
		select {
		case <-q.runChan:
			// 暂停
			q.SetStatus(STATUS_PAUSED)
			q.SetPauseAt(&currTime)
			q.LogInfo(fmt.Sprintf("协程[%d]停止调度！", i))
			// 判断是否已经获取到消息了，重新入队
			q.rePushMessages(messages, false)
			return
		default:
			// 设置心跳时间
			q.SetHeartbeatTime(&currTime)
			// 获取消息
			messages = []string{}
			messages := q.popMessages()
			if len(messages) > 0 {
				// 设置处理结果
				res := &QueueLastHandleResult{Time: currTime, Status: HANDLE_STATUS_PROGRESS}
				q.SetLastHandleResult(res)
				// 处理消息
				err := q.handleMessages(messages)
				if err != nil {
					// 记录异常日志
					q.LogError(fmt.Sprintf("协程[%d]处理消息异常！", i), err, map[string]any{"messages": messages})
					// 判断是否已经获取到消息了，重新入队
					q.rePushMessages(messages, true)
					res.Status = HANDLE_STATUS_FAILED
					res.Err = err
				} else {
					res.Status = HANDLE_STATUS_COMPLETED
				}
				// 结果记录
				endTime := time.Now().In(time.Local)
				res.Latency = util.MsTimeDiff(currTime, endTime)
				// 判断是否错误
				if res.Status == HANDLE_STATUS_FAILED {
					currHandleError := &QueueHandleError{Time: endTime, Latency: res.Latency, Err: res.Err}
					if len(q.handleErrors) >= 10 {
						q.handleErrors = append(q.handleErrors[1:], currHandleError)
					} else {
						if q.handleErrors == nil {
							q.handleErrors = make([]*QueueHandleError, 0, 10)
						}
						q.handleErrors = append(q.handleErrors, currHandleError)
					}
				}
				q.SetLastHandleResult(res)
			}
		}
	}
}

// 处理队列消息
func (q *Queue) handleMessages(messages []string) error {
	timeout := q.queueInstance.Timeout()
	done := make(chan bool)
	errChan := make(chan error)
	if timeout <= 0 {
		// 无超时时间
		func() {
			defer func() {
				if err := recover(); err != nil {
					errChan <- fmt.Errorf("panic: %v", err)
				}
			}()
			q.queueInstance.Handler(messages)
			done <- true
		}()
	} else {
		// 有超时时间
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		defer cancel()

		go func(ctx context.Context) {
			defer func() {
				if err := recover(); err != nil {
					errChan <- fmt.Errorf("panic: %v Stack trace: %s", err, string(debug.Stack()))
				}
			}()
			select {
			case <-ctx.Done():
				errChan <- fmt.Errorf("执行超时！")
				return
			default:
				q.queueInstance.Handler(messages)
				//panic("测试异常")
				done <- true
			}
		}(ctx)
	}

	select {
	case err := <-errChan:
		return err
	case <-done:
		return nil
	}
}

// 弹出队列消息
func (q *Queue) popMessages() []string {
	messages := []string{}
	messageNum := q.queueInstance.MessageNum()
	for i := 0; i < messageNum; i++ {
		message, err := redis.BRPop(context.Background(), 5*time.Second, q.queueInstance.Name())
		if err != nil && !errors.Is(err, goRedis.Nil) {
			log.Error(q.LogPrefix()+"获取队列消息异常！", err)
			return nil
		}
		if message != "" {
			messages = append(messages, message)
		}
	}
	return messages
}

// 消息尝试次数
var messageTries map[string]int

// 重新推送队列消息，并滞空消息变量
// isCheckTries: 是否检查尝试次数
func (q *Queue) rePushMessages(messages []string, isCheckTries bool) {
	// 检测尝试次数
	if limitTries := q.queueInstance.Tries(); limitTries > 0 && isCheckTries {
		newMessage := make([]string, 0, len(messages))
		if messageTries == nil {
			messageTries = make(map[string]int)
		}
		for _, message := range messages {
			triesKey := util.Md5(message)
			triesValue := messageTries[triesKey] + 1
			if triesValue > limitTries {
				// 超出尝试次数
				continue
			}
			messageTries[triesKey] = triesValue
			newMessage = append(newMessage, message)
			// fmt.Printf("重新入对：%v|%v|%v\r\n", triesKey, triesValue, message)
			// time.Sleep(5 * time.Second)
		}
		messages = newMessage
	}
	// 消息重新入对
	for _, message := range messages {
		if err := q.Push(message); err != nil {
			q.LogError("消息重新入队异常！", err, map[string]any{"message": message})
		}
	}
}

// 停止
func (q *Queue) Stop() {
	q.runMu.Lock()
	defer q.runMu.Unlock()
	select {
	case <-q.runChan:
		return
	default:
		close(q.runChan)
	}
	q.runWg.Wait()
}

// 日志
func (q *Queue) LogInfo(message string, withFields ...map[string]any) {
	log.Info(q.LogPrefix()+message, withFields...)
}

// 错误日志
func (q *Queue) LogError(message string, err any, withFields ...map[string]any) {
	log.Error(q.LogPrefix()+message, err, withFields...)
}
